#include<stdio.h>
#include<unistd.h>
#include<pthread.h>
#include<vector>
#include<semaphore.h>

using namespace std;
/*
 * 定义线程安全的队列
 *    环形队列（用数组模拟）
 *    线程安全：
 *       同步：信号量
 *       互斥： 信号量
 * */
#define CAPACITY 2
#define THREAD_COUNT 2

class RingQueue{
    public:
        RingQueue()
        :_vec(CAPACITY){
            _capacity = CAPACITY;
            sem_init(&_sem_lock, 0, 1);
            sem_init(&_cons_sem, 0, 0);
            sem_init(&_prod_sem, 0, CAPACITY);
            write_pos = 0;
            read_pos = 0;
        }
        ~RingQueue(){
            sem_destroy(&_sem_lock);
            sem_destroy(&_cons_sem);
            sem_destroy(&_prod_sem);
        }
        void push(int data){
            //获取生产者信号量
            sem_wait(&_prod_sem);
            //获取互斥锁
            sem_wait(&_sem_lock);
            printf("I am produce thread %p: I produce %d\n", pthread_self(), data);
            _vec[write_pos] = data;
            write_pos = (write_pos + 1) % _capacity;
            sem_post(&_sem_lock);
            //通知消费者进行消费
            sem_post(&_cons_sem);
        }
        void pop(){
            //获取消费者信号量
            sem_wait(&_cons_sem);
            sem_wait(&_sem_lock);
            int data = _vec[read_pos];
            printf("I am consum thread %p: I consum %d\n", pthread_self(), data);
            read_pos = (read_pos + 1) % _capacity; 
            sem_post(&_sem_lock);
            //通知生产者生产
            sem_post(&_prod_sem);
        }
    private:
        vector<int> _vec;
        size_t _capacity;
        //保证互斥的信号量
        sem_t _sem_lock;  
        sem_t _cons_sem;  //消费者的信号量
        sem_t _prod_sem;  //生产者的信号量

        int write_pos;
        int read_pos;
};
int g_data = 0;
sem_t g_lock;
void* prod_thread_start(void* arg){
    RingQueue* rq = (RingQueue*)arg;
    while(1){
        sem_wait(&g_lock);
        rq->push(g_data);
        g_data++;
        sem_post(&g_lock);
    }
}
void* cons_thread_start(void* arg){
    RingQueue* rq = (RingQueue*)arg;
    while(1){
        rq->pop();
    }
}
int main(){
    RingQueue* rq = new RingQueue();
    //保证多个生产者互斥访问
    sem_init(&g_lock, 0, 1);

    pthread_t cons[THREAD_COUNT], prod[THREAD_COUNT];
    for(int i = 0; i < THREAD_COUNT; ++i){
        int ret = pthread_create(&prod[i], NULL, prod_thread_start, (void*)rq);
        if(ret < 0){
            perror("pthread_create");
        }
        ret = pthread_create(&cons[i], NULL, cons_thread_start, (void*)rq);
        if(ret < 0){
            perror("pthread_create");
        }
    }
    for(int i = 0; i < THREAD_COUNT; ++i){
        pthread_join(prod[i], NULL);
        pthread_join(cons[i], NULL);
    }
    sem_destroy(&g_lock);
    delete rq;
    return 0;
}
